package com.bbx.flink.table_api_or_sql;
import com.bbx.flink.demo.entity.UserAction;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

import static org.apache.flink.table.api.Expressions.$;

/**
 * Flink table api   processing time   watermark demo
 * 通过table  api 创建 Processing  time watermark
 * The processing time attribute is defined with the property during schema definition.
 * The time attribute must only extend the physical schema by an additional logical field.
 * Thus, it is only definable at the end of the schema definition..proctime
 *
 */
@Slf4j
public class TimeAttributeDataStreamProcessingTimeDemo {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        DataStream<UserAction> dataStream = env.socketTextStream("114.116.104.74", 10003)
                .map(i -> {
                    log.info("~~~~~~~~~[{}]", i);
                    String[] split = i.split(",");
                    UserAction userAction = new UserAction(split[0], split[1], Long.parseLong(split[2]));
                    log.info("!!!!![{}]", userAction.toString());
                    return userAction;
                });

        //Thus, it is only definable at the end of the schema definition..proctime
        //不太理解，$("active_time_time").proctime() 放在中间也是可以的
        Table table = tableEnv.fromDataStream(dataStream,
                $("id"), $("name") ,$("active_time") ,$("active_time_time").proctime());
//        Table table = tableEnv.fromDataStream(dataStream,
//                $("id"), $("name"),$("active_time_time").proctime() ,$("active_time") );
        tableEnv.toAppendStream(table,Row.class).print();

        env.execute();

    }

}
